package com.ww.springboot.boot.rxjava.rxjava2;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import rx.Emitter;
import rx.Observable;
import rx.functions.Action0;
import rx.functions.Action1;

@Slf4j
public class RxJavaObserverDemo3 {
    /**
     * 演示RxJava中的不完整观察者
     */
    @Test
    public void rxJavaActionDemo() {
        //被观察者（主题）
        Observable observable = Observable.create(
                (Action1<Emitter<String>>) emitter -> {
                    emitter.onNext("apple");
                    emitter.onNext("banana");
                    emitter.onNext("pear");
                    emitter.onCompleted();
                }, Emitter.BackpressureMode.NONE);

        Action1<String> onNextAction = s -> log.info(s);
        Action1<Throwable> onErrorAction = throwable -> log.info("onError,Error Info is:" + throwable.getMessage());
        Action0 onCompletedAction = () -> log.info("onCompleted");
        log.info("第1次订阅：");
        //根据onNextAction来定义onNext()
        observable.subscribe(onNextAction);

        log.info("第2次订阅：");
        //根据onNextAction来定义onNext()，根据onErrorAction来定义onError()
        observable.subscribe(onNextAction, onErrorAction);

        log.info("第3次订阅：");
        //根据onNextAction来定义onNext()，根据onErrorAction来定义onError()
        //根据onCompletedAction来定义onCompleted()
        observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
    }
}